/*
 * enc_istener.cpp
 *
 *  Created on: 2018年11月6日
 *  Author: Degang.Wu (windragon@126.com)
 *  Desc:
 */

#include "enc_listener.h"

namespace hiapp
{

int Listener::WaitSignal(pthread_mutex_t* _lock, pthread_cond_t* _cond, int msec)
{
	struct timeval now_time;
	struct timespec ts;
	gettimeofday(&now_time, NULL);
	ts.tv_sec = now_time.tv_sec + msec / 1000;
	ts.tv_nsec = now_time.tv_usec * 1000 + (msec % 1000) * 1000000;
	if (ts.tv_nsec >= 1000000000) {
		ts.tv_sec += 1;
		ts.tv_nsec -= 1000000000;
	}

	pthread_mutex_lock(_lock);
	if (msec > 0) {
		pthread_cond_timedwait(_cond, _lock, &ts);
	} else {
		pthread_cond_wait(_cond, _lock);
	}
	pthread_mutex_unlock(_lock);
	return 0;
}

int Listener::NotifySignal(pthread_mutex_t* _lock, pthread_cond_t* _cond)
{
	pthread_mutex_lock(_lock);
	pthread_cond_signal(_cond);
	pthread_mutex_unlock(_lock);
	return 0;
}

Listener::Listener(U16 _port)
	: u::sys::Thread("Listener")
	, th_listen(0)
	, nPort(_port)
{
	// TODO Auto-generated constructor stub

}

Listener::~Listener()
{
	// TODO Auto-generated destructor stub
}


void Listener::run()
{
	pthread_create(&th_listen, NULL, listen_thread, this);

	Delay(1000);

	if (running())
	{
		onRun();
	}

	if (th_listen)
	{
		pthread_join(th_listen, NULL);
		th_listen = 0;
	}
	BW_INFO("[%s] listener quit", enc_is_master() ? "MASTER" : "SLAVE");
}


void* Listener::listen_thread(void* param)
{
	Listener* pSlave = (Listener*) param;
	if (pSlave)
	{
		pSlave->onListen();
	}
	pthread_exit(NULL);
}

void Listener::onListen()
{
	while (m_msgSrv.listenx(nPort) < 0 && running()) {
		Delay(2000);
		BW_INFO("Listen %d failed, retry after 2 seconds...", nPort);
	}

	BW_INFO("[%s] Listen %d success...", enc_is_master() ? "MASTER" : "SLAVE", nPort);

	fd_set rfd;
	int nfds;
	struct sockaddr_in client_addr;
	int serversock = m_msgSrv.socketx();
	int clientsock = -1;
	while (running())
	{
		struct timeval timeout = {1, 0};
		FD_ZERO(&rfd);
		FD_SET(serversock, &rfd);
		nfds = select(serversock+1, &rfd, NULL, NULL, &timeout);

		if ( 0 == nfds ) {
			continue;
		}
		else if ( nfds > 0)
		{
			FD_CLR(serversock, &rfd);
			socklen_t clientlen = sizeof(sockaddr_in);
			memset(&client_addr, 0, sizeof(client_addr));
			if ( (clientsock = accept(serversock, (struct sockaddr *)&client_addr, &clientlen)) < 0 ) {
				break;
			}

			//BW_INFO("connect from [%s:%d]", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));

			struct timeval timeout = {1, 0};
			setsockopt(clientsock,  SOL_SOCKET, SO_RCVTIMEO, (char*)&timeout, sizeof(struct timeval));

			pMSG_HEAD msghead = (pMSG_HEAD)msgbuf;
			int ret = ::recv(clientsock, (char*)msghead, sizeof(stMSG_HEAD), 0);
			if (ret != sizeof(stMSG_HEAD)) {
				continue;
			}

			NTOH_MSGHEAD(msghead);
//			BW_INFO("msg type=%d, len=%d", msghead->msg_type, msghead->msg_len);

			U32 want = msghead->msg_len;
		    int result = 0;
		    U32 recvlen = 0;

		    if (want > (sizeof(msgbuf) - sizeof(stMSG_HEAD)))
		    {
		    	BW_INFO("msg len is too large");
		    	continue;
		    }

		    pU8 buf = msgbuf + sizeof(stMSG_HEAD);
		    memset(buf, 0, sizeof(msgbuf) - sizeof(stMSG_HEAD));
		    while ( recvlen < want ) {
		    	result = ::recv(clientsock, (char*)&buf[recvlen], (want - recvlen), 0);
		    	if ( result <= 0 ) {
		    		break;
		    	}
		    	recvlen += result;
		    }

		    // process buf
		    process(msgbuf, clientsock);

		    ::close(clientsock);
		}
	}

	BW_INFO("Listener stop ...");
}

void Listener::stop()
{
	tryterminate();
	m_msgSrv.closex();
	WakeUp();

	onStop();

	terminate();
}

} /* namespace hiapp */
